package lv.solcraft.replication.client;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;

import lv.solcraft.replication.client.logging.ClientLogger;
import lv.solcraft.replication.common.OperDTO;
import lv.solcraft.replication.common.OperStateDTO;
import lv.solcraft.replication.common.ReplicationConst;
import lv.solcraft.replication.common.util.ShutdownEvent;

import org.apache.log4j.Logger;
import org.springframework.context.ApplicationEvent;
import org.springframework.context.ApplicationListener;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.support.JdbcDaoSupport;
import org.springframework.jdbc.datasource.DataSourceUtils;
import org.springframework.jdbc.support.JdbcUtils;
import org.springframework.jdbc.support.rowset.SqlRowSet;
import org.springframework.util.Assert;

public class OperExecutorImpl extends JdbcDaoSupport implements OperExecutor, ApplicationListener {

    private static final String SCR_RECORD_ID = "SCR_RECORD_ID";

    private static final String SCR_SITE_ID = "SCR_SITE_ID";

    private static Logger log = Logger.getLogger(OperExecutorImpl.class);

    private static final int MARKER_HAS_ERRORS = -1;

    private ClientConfig client;

    private OperExecutorSqlHolder sqlHolder;

    private ClientLogger clientLogger;

    private volatile boolean stop;

    public List<OperStateDTO> execute(List<OperDTO> opers) {
        stop = false;
        List<OperStateDTO> states = new ArrayList<OperStateDTO>(opers.size());
        int index = 0;
        for (OperDTO oper : opers) {
            if (stop) {
                throw new RuntimeException(
                        "Forcing transaction rollback in oper executor, because shutdown event was received.");
            }
            if (oper.marker != MARKER_HAS_ERRORS) {
                OperStateDTO state = executeOper(oper);
                // if ERROR state, then update
                // all other opers for same record to ERROR
                if (state.getState().equals(ReplicationConst.STATE_ERROR)) {
                    setStatesToError(opers, index, oper, states);
                }
                states.add(state);
            } else {
                // oper already marked for error
                OperStateDTO state = new OperStateDTO(oper, ReplicationConst.STATE_ERROR, "Record already has errors");
                states.add(state);
            }
            index++;
        }
        return states;
    }

    private OperStateDTO executeOper(OperDTO oper) {
        OperStateDTO state = null;
        try {
            if (useStoredProc(oper.getTable())) {
                state = executeOperWithStoredProc(oper);
            } else {
                state = executeOperDirect(oper);
            }
        } catch (DataAccessException e) {
            log.error(e);
            clientLogger.errorLocal("Error while executing oper " + oper.getSysOperId() + ": "
                    + e.getMostSpecificCause().getMessage());
            return new OperStateDTO(oper, ReplicationConst.STATE_ERROR, e.getMostSpecificCause().getMessage());
        }
        if (oper.isTypeUpdate()) {
            try {
                executeBlobFieldsUpdate(oper);
            } catch (DataAccessException e) {
                log.error(e);
                clientLogger.errorLocal("Error while executing oper blob fields " + oper.getSysOperId() + ": "
                        + e.getMessage());
            }
        }
        return state;
    }

    private void executeBlobFieldsUpdate(OperDTO oper) {
        for (Entry<String, byte[]> entry : oper.getBlobFields().entrySet()) {
            getJdbcTemplate().update(
                    "UPDATE " + oper.getTable() + " SET " + entry.getKey()
                            + " = ? WHERE SCR_SITE_ID = ? AND SCR_RECORD_ID = ? ",
                    new Object[] { entry.getValue(), oper.getRecordSiteId(), oper.getRecordId() });
        }

    }

    private void setStatesToError(List<OperDTO> opers, int index, OperDTO oper, List<OperStateDTO> states) {
        for (int i = index + 1; i < opers.size(); i++) {
            OperDTO otherOper = opers.get(i);
            if (otherOper.getRecordId() == oper.getRecordId() && otherOper.getRecordSiteId() == oper.getRecordSiteId()) {
                otherOper.marker = MARKER_HAS_ERRORS;
            }
        }
    }

    private OperStateDTO executeOperWithStoredProc(OperDTO oper) {
        if (oper.isTypeDelete()) {
            return executeDeleteOperWithStoredProc(oper);
        } else if (oper.isTypeInsert()) {
            return executeInsertOperWithStoredProc(oper);
        } else if (oper.isTypeUpdate()) {
            return executeUpdateOperWithStoredProc(oper);
        } else {
            throw new IllegalArgumentException("Invalid oper type " + oper.getType());
        }
    }

    private OperStateDTO executeUpdateOperWithStoredProc(final OperDTO oper) {
        String sql = sqlHolder.getStoredProcStatement(oper.getTable(), oper.getType());
        List<String> paramList = sqlHolder.getStoredProcParams(oper.getTable(), oper.getType());
        String recordSql = sqlHolder.getCurrRecordStatement(oper.getTable());
        if (paramList.isEmpty()) {
            return new OperStateDTO(oper, ReplicationConst.STATE_ERROR, "Procedure does not exists for table "
                    + oper.getTable() + ", type " + oper.getType());
        }
        final String[] params = new String[paramList.size()];
        // set current field values
        SqlRowSet row = getJdbcTemplate().queryForRowSet(recordSql,
                new Object[] { oper.getRecordSiteId(), oper.getRecordId() });
        if (row.next()) {
            for (String field : row.getMetaData().getColumnNames()) {
                field = field.trim();
                int index = paramList.indexOf(field);
                if (index >= 0) {
                    String value = row.getString(field);
                    params[index] = value;
                }
            }
        }
        // set fields values from oper
        setScrSiteAndRecordParamValues(oper, paramList, params);
        for (OperDTO.Field field : oper.getFields()) {
            int index = paramList.indexOf(field.getName());
            if (index >= 0) {
                params[index] = field.getValue();
            }
        }
        OperStateDTO state = executeStoredProc(oper, sql, params);
        return state;
    }

    private boolean recordExists(final OperDTO oper) {
        List result = getJdbcTemplate().queryForList(
                "SELECT 0 FROM " + oper.getTable() + " WHERE SCR_RECORD_ID = ? AND SCR_SITE_ID = ?",
                new Object[] { oper.getRecordId(), oper.getRecordSiteId() }, Integer.class);
        return (!(result.size() == 0));
    }

    private OperStateDTO executeInsertOperWithStoredProc(final OperDTO oper) {
        if (recordExists(oper)) {
            return new OperStateDTO(oper, ReplicationConst.STATE_IGNORED, "Record already exists, insert ignored");
        }
        String sql = sqlHolder.getStoredProcStatement(oper.getTable(), oper.getType());
        List<String> paramList = sqlHolder.getStoredProcParams(oper.getTable(), oper.getType());
        if (paramList.isEmpty()) {
            return new OperStateDTO(oper, ReplicationConst.STATE_ERROR, "Procedure does not exists for table "
                    + oper.getTable() + ", type " + oper.getType());
        }
        String[] params = new String[paramList.size()];
        setScrSiteAndRecordParamValues(oper, paramList, params);
        for (OperDTO.Field field : oper.getFields()) {
            int index = paramList.indexOf(field.getName());
            if (index >= 0) {
                params[index] = field.getValue();
            }
        }
        OperStateDTO state = executeStoredProc(oper, sql, params);
        return state;
    }

    private OperStateDTO executeDeleteOperWithStoredProc(final OperDTO oper) {
        String sql = sqlHolder.getStoredProcStatement(oper.getTable(), oper.getType());
        List<String> paramList = sqlHolder.getStoredProcParams(oper.getTable(), oper.getType());
        if (paramList.isEmpty()) {
            return new OperStateDTO(oper, ReplicationConst.STATE_ERROR, "Procedure does not exists for table "
                    + oper.getTable() + ", type " + oper.getType());
        }
        String[] params = new String[2];
        setScrSiteAndRecordParamValues(oper, paramList, params);
        OperStateDTO state = this.executeStoredProc(oper, sql, params);
        return state;
    }

    private void setScrSiteAndRecordParamValues(OperDTO oper, List<String> paramList, String[] params) {
        params[paramList.indexOf(SCR_SITE_ID)] = Integer.toString(oper.getRecordSiteId());
        params[paramList.indexOf(SCR_RECORD_ID)] = Integer.toString(oper.getRecordId());
    }

    private OperStateDTO executeStoredProc(OperDTO oper, String sql, String[] params) {
        Connection con = null;
        PreparedStatement pstmt = null;
        ResultSet rs = null;
        try {
            con = DataSourceUtils.getConnection(getDataSource());
            pstmt = con.prepareStatement(sql);
            for (int i = 0; i < params.length; i++) {
                pstmt.setString(i + 1, params[i]);
            }
            rs = pstmt.executeQuery();
            if (!rs.next()) {
                throw new StoredProceExecException("Stored procedure for table " + oper.getTable() + " and type "
                        + oper.getType() + " doesnt return result");
            }

            String state = rs.getString(1);
            Assert.hasText(state, "Stored procedure returned empty state, sql: " + sql);
            String msg = rs.getString(2);
            if (msg == null) {
                msg = "";
            }
            return new OperStateDTO(oper, state, msg);
        } catch (SQLException ex) {
            throw new StoredProceExecException("Error while exec proc, " + ex.getMessage() + ", " + sql);

        } finally {
            JdbcUtils.closeResultSet(rs);
            JdbcUtils.closeStatement(pstmt);
            if (con != null) {
                DataSourceUtils.releaseConnection(con, getDataSource());
            }
        }
    }

    private OperStateDTO executeOperDirect(OperDTO oper) {
        if (oper.isTypeDelete()) {
            return executeDeleteOperDirect(oper);
        } else if (oper.isTypeInsert()) {
            return executeInsertOperDirect(oper);
        } else if (oper.isTypeUpdate()) {
            return executeUpdateOperDirect(oper);
        } else {
            throw new IllegalArgumentException("Invalid oper type " + oper.getType());
        }
    }

    private OperStateDTO executeUpdateOperDirect(OperDTO oper) {
        List<String> values = new ArrayList<String>();
        StringBuffer sql = new StringBuffer();
        sql.append("UPDATE ").append(oper.getTable()).append(" SET ");
        for (int i = 0; i < oper.getFields().size(); i++) {
            // check if field is in table
            if (sqlHolder.isFieldInTable(oper.getTable(), oper.getFields().get(i).getName())) {
                if (values.size() > 0) {
                    sql.append(", ");
                }
                sql.append(oper.getFields().get(i).getName()).append(" = ? ");
                values.add(oper.getFields().get(i).getValue());
            }
        }
        if (values.size() == 0) {
            return new OperStateDTO(oper, ReplicationConst.STATE_IGNORED, "No fields to update");
        }
        sql.append(" WHERE SCR_SITE_ID = ? AND SCR_RECORD_ID = ?");
        values.add(Integer.toString(oper.getRecordSiteId()));
        values.add(Integer.toString(oper.getRecordId()));
        getJdbcTemplate().update(sql.toString(), values.toArray());
        return new OperStateDTO(oper, ReplicationConst.STATE_COMMITED, "");
    }

    private OperStateDTO executeInsertOperDirect(OperDTO oper) {
        if (recordExists(oper)) {
            return new OperStateDTO(oper, ReplicationConst.STATE_IGNORED, "Record already exists, insert ignored");
        }
        List<String> values = new ArrayList<String>();
        StringBuffer sql = new StringBuffer();
        StringBuffer sql2 = new StringBuffer();
        sql.append("INSERT INTO ").append(oper.getTable()).append(" ( ");
        sql2.append(" VALUES (");
        for (int i = 0; i < oper.getFields().size(); i++) {
            // check if field is in table
            if (sqlHolder.isFieldInTable(oper.getTable(), oper.getFields().get(i).getName())) {
                if (values.size() > 0) {
                    sql.append(", ");
                    sql2.append(", ");
                }
                sql.append(oper.getFields().get(i).getName());
                sql2.append("?");
                values.add(oper.getFields().get(i).getValue());
            }
        }

        if (values.size() == 0) {
            return new OperStateDTO(oper, ReplicationConst.STATE_IGNORED, "No fields in insert statement");
        }
        sql.append(", SCR_SITE_ID, SCR_RECORD_ID) ");
        sql2.append(", ?, ?) ");
        sql.append(sql2);
        values.add(Integer.toString(oper.getRecordSiteId()));
        values.add(Integer.toString(oper.getRecordId()));
        getJdbcTemplate().update(sql.toString(), values.toArray());
        return new OperStateDTO(oper, ReplicationConst.STATE_COMMITED, "");

    }

    private OperStateDTO executeDeleteOperDirect(OperDTO oper) {
        getJdbcTemplate().update(
                "DELETE FROM " + oper.getTable() + " WHERE SCR_SITE_ID = ? " + " AND SCR_RECORD_ID = ?",
                new Object[] { oper.getRecordSiteId(), oper.getRecordId() });
        return new OperStateDTO(oper, ReplicationConst.STATE_COMMITED, "");
    }

    public void setClient(ClientConfig client) {
        this.client = client;
    }

    public void setSqlHolder(OperExecutorSqlHolder sqlHolder) {
        this.sqlHolder = sqlHolder;
    }

    public void setClientLogger(ClientLogger clientLogger) {
        this.clientLogger = clientLogger;
    }

    public void onApplicationEvent(ApplicationEvent event) {
        if (event instanceof ShutdownEvent) {
            log.info("Received shutdown event");
            stop = true;
        }
    }

    private boolean useStoredProc(String table) {
        return client.getSettings().useStoredProcsForTable(table);
    }

    public static class StoredProceExecException extends DataAccessException {
        public StoredProceExecException(String arg0) {
            super(arg0);
        }

        public StoredProceExecException(String arg0, Throwable arg1) {
            super(arg0, arg1);
        }
    }
}
